package com.share.device.receiver;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.share.common.rabbit.MqConstant.MqConst;
import com.share.common.rabbit.model.CanalSyncMessage;
import com.share.device.receiver.processSync.ProcessRedisSync;
import com.share.device.receiver.processSync.ProcessStationRedisSync;
import com.share.system.api.service.RegionRedisService;
import com.share.system.api.domain.Region;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@Component
@Slf4j
@RefreshScope
public class ReceiverDevice {

    @Autowired
    private ProcessRedisSync processRedisSync;

    @Autowired
    private ProcessStationRedisSync processStationRedisSync;

    /**
     * 1.消费者接收消息
     * 2.获取消息头的初始重试次数，如果同步redis成功手动ack，
     * 如果没有成功，则手动nack，重新进入队列进行重试
     * 3.如果超过50次重试，则进入死信队列
     *
     * 1.创建死信交换机和队列和路由
     * 2.业务队列绑定死信交换机（无需在rabbitmqConfig中配置,直接在业务队列上绑定）
     */



    /**
     * 接收redis同步消息
     * 1.处理可能出现的异常，防止消息丢失
     * @param message
     */
    @RabbitListener(queues = MqConst.QUEUE_REDIS_SYNC, containerFactory = "customContainerFactory")
    public void receiveRigion(String message, Channel channel,
                              @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag
//                            ,@Header(MqConst.RETRY_COUNT_HEADER) Integer retryCount
                            ) {

        try {
            log.info("接收到Redis同步消息: {}", message);
            /**
             * 可能存在转换异常，这里使用try-catch
             */
            CanalSyncMessage canalSyncMessage = JSON.parseObject(message, CanalSyncMessage.class);
            log.info("接收实体类: {}", JSON.toJSON(canalSyncMessage));
            boolean processResult = false;
            String tableName = canalSyncMessage.getTable();
            if("region".equals(tableName)){
                processResult = processRedisSync.processRedisSync(canalSyncMessage);
            }else if ("station".equalsIgnoreCase(tableName)) {
                processResult = processStationRedisSync.processStationRedisSync(canalSyncMessage);
            } else {
                log.warn("未知的表类型: {}", tableName);
                processResult = true; // 对于未知表直接确认消息
            }
            if (processResult) {
                // 手动确认消息
                channel.basicAck(deliveryTag, false);
                log.info("消息处理成功，已确认: {}", deliveryTag);
            } else {
                // 处理失败，根据重试次数决定是否进入死信队列
                handleProcessFailure(channel, deliveryTag, message);
            }

        } catch (Exception e) {
            log.error("处理消息失败: {}", message, e);
            handleProcessFailure(channel, deliveryTag, message);
        }
    }

    /**
     * 目前消息接收失败之后直接进入死信队列
     *
     * 可设置最大重试次数，默认50，使用try-catch,加延迟重新发生队列，在消费者里面判断是否达到最大重试次数
     * 处理消息处理失败的情况
     */
    private void handleProcessFailure(Channel channel, Long deliveryTag,
                                      String message) {
        try {
//            if (retryCount < MqConst.MAX_RETRY_COUNT) {
//                // 未达到最大重试次数，重新入队
//                channel.basicNack(deliveryTag, false, true);
//                log.info("消息处理失败，重新入队，重试次数: {}/{}", retryCount + 1, MqConst.MAX_RETRY_COUNT);
//            } else {
            // 达到最大重试次数，进入死信队列
            channel.basicNack(deliveryTag, false, false);
            log.info("消息重试{}次后仍失败，进入死信队列: {}", MqConst.MAX_RETRY_COUNT, deliveryTag);
//            }
        } catch (IOException ioException) {
            log.error("消息处理失败: {}", deliveryTag, ioException);
        }
    }

}
